package com.fomky.queried.plugin.mysql;

import com.fomky.queried.common.connect.RowData;
import com.fomky.queried.common.connect.RowDataList;
import com.fomky.queried.common.connect.SchemaColumn;
import com.fomky.queried.plugin.common.AbstractConnectPlugin;
import com.fomky.queried.plugin.common.ConnectConfig;
import com.fomky.queried.plugin.common.ConnectPlugin;
import io.asyncer.r2dbc.mysql.MySqlConnectionConfiguration;
import io.asyncer.r2dbc.mysql.MySqlConnectionFactory;
import io.asyncer.r2dbc.mysql.api.MySqlConnection;
import io.asyncer.r2dbc.mysql.api.MySqlResult;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Result;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;

@Slf4j
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@EqualsAndHashCode(callSuper = true)
public class MySQLConnectPlugin extends AbstractConnectPlugin {

    protected ConnectConfig config;

    protected MySqlConnectionFactory connectionFactory;

    protected Mono<MySqlConnection> connectionMono;

    public MySQLConnectPlugin(ConnectConfig config) {
        this.config = config;
    }

    @Override
    public String name() {
        return "mysql";
    }

    @Override
    public ConnectPlugin init(ConnectConfig config) {

        // Notice: the query string must be URL encoded
        // r2dbc:pool:mysql://root:database-password-in-here@127.0.0.1:3306/r2dbc
        MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builder()
                .connectionTimeZone(config.getConnectionTimeZone())
                .host(config.getHost())
                .port(config.getPort())
                .user(config.getUser())
                .createDatabaseIfNotExist(false)
                .password(config.getPassword())
                .build();
        connectionFactory = MySqlConnectionFactory.from(configuration);
        // Creating a Mono using Project Reactor
        connectionMono = Mono.from(connectionFactory.create());
        return this;
    }

    public Mono<MySqlConnection> connection() {
        if (connectionMono == null) {
            this.init(config);
        }
        return connectionMono;
    }

    @Override
    public RowDataList catalogs() {
        List<RowData> data = new ArrayList<>();
        data.add(RowData.builder()
                .columns(List.of(SchemaColumn.builder().name("catalog").type("string").build()))
                .data(Map.of("catalog", "mysql"))
                .build());
        return RowDataList.of(data);
    }

    @Override
    public RowDataList databases(String catalog) {
        return executeSql("show databases");
    }

    public RowDataList executeSql(String sql) {
        Publisher<? extends Result> publisher =
                this.execute(connection ->
                        connection.createStatement(sql).execute());
        Result publisherResult = Mono.from(publisher).block();
        if (publisherResult instanceof MySqlResult result) {
            List<RowData> data = result.map(new ResultMapFunction()).buffer().blockLast();
            if (Objects.nonNull(data)) {
                return RowDataList.of(data);
            }
        }
        return null;
    }

    public <T> T execute(Function<Connection, T> function) {
        MySqlConnection connection = null;
        try {
            connection = connection().blockOptional().get();
            return function.apply(connection);
        } catch (Exception e) {
            log.error("Error: {}", e.getMessage(), e);
            throw e;
        } finally {
            if (Objects.nonNull(connection)) {
                connection.close();
            }
        }
    }


    @Override
    public RowDataList tables(String catalog, String database) {
        return executeSql("show tables from " + database);
    }

    @Override
    public RowDataList tableInfo(String catalog, String database, String table) {
        return executeSql(String.format("desc %s.%s",database,table));
    }

    @Override
    public RowDataList execute(String requestSql) {
        return executeSql(requestSql);
    }
}
